package com.wis.bolt.mpg;

import com.wis.TopologyDrive;
import org.apache.commons.lang.StringUtils;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.LinkedList;
import java.util.List;
import java.util.Map;

/**
 * @Description: traffic_wis    将三个队列(集合)中的数值进行一一匹配
 * @author: fan
 * @Date: Created in 2018/12/17 14:02
 * @Modified By:
 */
public class MPGJiHeDanJiHeBolt extends BaseRichBolt {
    private OutputCollector collector;
    private static String typeProduce;

    @Override
    public void prepare(Map map, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
        typeProduce = map.get(TopologyDrive.TYPE_PRODUCE).toString();
    }

    @Override
    public void execute(Tuple tuple) {
        String key = tuple.getStringByField("key");
        String value = tuple.getStringByField("yaosu");
        String id = tuple.getStringByField("id");
        String xt = tuple.getStringByField("xt");
        // System.out.println(key + "------" + value.split("=", -1).length + "----" + id.split("=", -1).length + "---" + xt.split("=", -1).length);
        try {
            if (StringUtils.isNotBlank(value) && value.length() > 4) {
                String[] ids = id.split("=", -1);
                String[] xts = xt.split("=", -1);
                // 卡值的时候，前面多加了一个逗号，温度会有负值，所以就按这样切割
                String[] hang = value.split("=", -1);
                String[][] arrs = new String[hang.length][hang[0].split(",", -1).length];
                for (int i = 0; i < hang.length; i++) {// 有多少行
                    arrs[i] = hang[i].split(",", -1);
                }
                for (int i = 0; i < arrs[0].length; i++) {// 有多少列
                    List list = new LinkedList();// TODO 使用redis进行控制

                    for (int j = 0; j < arrs.length; j++) {
                        if (StringUtils.isNotBlank(arrs[j][i]) && !arrs[j][i].contains("-999.9")) {
                            list.add(arrs[j][i] + "_" + xts[j].split(",", -1)[i] + "_" + ids[j]);
                        }
                    }
                    if (list.size() > 1)
                        if (typeProduce.equals("1")) {
                            this.collector.emit(new Values(key.replace("sc", "00"), StringUtils.join(list, ",")));
                        } else if (typeProduce.equals("2")) {
                            this.collector.emit(new Values(key.replace("sc", (i + 1) + "0"), StringUtils.join(list, ",")));
                        } else if (typeProduce.equals("3")) {
                            if (i < 9) {
                                this.collector.emit(new Values(key.replace("sc", "0" + (i + 1)), StringUtils.join(list, ",")));
                            } else {
                                this.collector.emit(new Values(key.replace("sc", (i + 1) + ""), StringUtils.join(list, ",")));
                            }
                        }
                }
            } else if (value.equals("jies")) {
                if (typeProduce.equals("1")) {
                    this.collector.emit(new Values(key.replace("sc", "00"), "over"));
                } else if (typeProduce.equals("2")) {
                    this.collector.emit(new Values(key.replace("sc", "40"), "over"));
                } else if (typeProduce.equals("3")) {
                    this.collector.emit(new Values(key.replace("sc", "80"), "over"));
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            this.collector.ack(tuple);
        }

    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("key", "value"));
    }

}

